package com.fanxuankai.boot.canal.listener;

import cn.hutool.core.text.StrPool;
import com.fanxuankai.boot.canal.listener.annotation.CanalListener;
import com.fanxuankai.boot.canal.util.DomainConverter;
import com.fanxuankai.boot.canal.util.Pair;
import com.fanxuankai.boot.canal.util.Topic;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import javax.annotation.Resource;
import java.util.*;

import static com.fanxuankai.boot.canal.listener.EventType.*;

/**
 * @author fanxuankai
 */
public class MessageListener implements ApplicationContextAware {
    @Resource
    private CanalTableCache canalTableCache;
    protected final Map<String, CanalListenerDefinition> byTopic = new HashMap<>(16);
    protected final List<CanalListenerDefinition> definitions = new ArrayList<>(16);

    public void onMessage(String topic, String data) {
        int i = topic.lastIndexOf(StrPool.UNDERLINE);
        String rawTopic = topic.substring(0, i);
        EventType eventType = valueOf(topic.substring(i + 1));
        CanalListenerDefinition definition = byTopic.get(rawTopic);
        assert definition != null;
        if (eventType == INSERT && definition.getInsertConsumer() != null) {
            definition.getInsertConsumer().accept(DomainConverter.of(data, definition.getEntityClass()));
        } else if (eventType == UPDATE && definition.getUpdateConsumer() != null) {
            Pair<Object, Object> pair = DomainConverter.pairOf(data, definition.getEntityClass());
            definition.getUpdateConsumer().accept(pair.getKey(), pair.getValue());
        } else if (eventType == DELETE && definition.getDeleteConsumer() != null) {
            definition.getDeleteConsumer().accept(DomainConverter.of(data, definition.getEntityClass()));
        }
    }

    public Set<String> getTopics() {
        Set<String> topics = new HashSet<>();
        for (CanalListenerDefinition definition : definitions) {
            if (definition.getInsertConsumer() != null) {
                topics.add(Topic.customWithoutPrefix(definition.getTopic(), INSERT.name()));
            }
            if (definition.getUpdateConsumer() != null) {
                topics.add(Topic.customWithoutPrefix(definition.getTopic(), UPDATE.name()));
            }
            if (definition.getDeleteConsumer() != null) {
                topics.add(Topic.customWithoutPrefix(definition.getTopic(), DELETE.name()));
            }
        }
        return topics;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        for (Object canalListener : applicationContext.getBeansWithAnnotation(CanalListener.class).values()) {
            CanalListenerDefinition definition = new CanalListenerDefinition(canalListener, canalTableCache);
            definitions.add(definition);
            byTopic.put(definition.getTopic(), definition);
        }
    }
}
